package com.hzg.stream.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;

/**
 * @Author: huangzhigao
 * @Date: 2022/4/20 16:42
 */
public class OperatorStateFunction implements MapFunction<String, String>, CheckpointedFunction {

    private ValueState<String> valueState;
    private ListState<String> listState;

    @Override
    public String map(String value) throws Exception {
        return null;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        //制作checkpoint的时候清理状态
        valueState.clear();

    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        //注册状态
        valueState = context.getKeyedStateStore().getState(new ValueStateDescriptor<String>("value", String.class));
        listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("listState", String.class));
    }
}
